package com.example.test;

import java.io.IOException;
import java.io.PrintStream;
import java.util.Map;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.GenericMessage;

import com.example.model.User;
import com.example.utils.MapUtils;

@SpringBootApplication
public class DemoKafkaSpringApplication implements CommandLineRunner {
	private final static String topics = "1234567890";

	// @KafkaListeners({@KafkaListener(topics=topics,containerFactory="kafkaBatchListenerContainerFactory")})
	// void TestKafkakaListener(@Payload List<String> bytesUsers ){
	// System.err.println(bytesUsers.size());
	//// System.err.println(new String(bytesUsers.get(0)));
	// }
	@KafkaHandler
	@KafkaListener(topics = topics)
	void TestKafkakaListener(@Payload User bytesUser, @Headers Map<String, Object> headers) {
		System.err.println(bytesUser);
		System.err.println(headers);
	}

	public static void main(String[] args) throws IOException {
		SpringApplication.run(DemoKafkaSpringApplication.class, args);
	}

	@Autowired
	KafkaTemplate<String, Object> kafkaTemplate;

	@Override
	public void run(String... args) throws Exception {
		User user = new User();
		user.setLeavl("admin");
		
		for (int i = 0; i < 100; i++) {
			user.setUserid(i);
			user.setUsername("张三" + i);
			
			Map<String, Object> headers = MapUtils.<String, Object>n().a(KafkaHeaders.TOPIC, topics).a(KafkaHeaders.PARTITION_ID, 0).a(KafkaHeaders.MESSAGE_KEY, user.toString()).to();
			kafkaTemplate.send(new GenericMessage<User>(user, headers));
		}
		kafkaTemplate.flush();
		kafkaTemplate.execute(producer -> {
			ProducerRecord<String, Object> record = new ProducerRecord<>("rrrrrrrrrrrr2222", 6, "ee", "eee");
			producer.send(record);
			return null;
		});
		kafkaTemplate.flush();
	}

}
